package com.hopu.user.time;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.time.TimeSex;
import com.hopu.bean.time.TimeTotal;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class TimeSexAna {

    public static void main(String[] args) {

        //加载日志文件，生成RDD
        SparkConf conf = new SparkConf().setAppName("timeSexAna").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        //把原始日志数据转换成实体类的RDD
        JavaPairRDD<TimeSex, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String hour = json.getJSONObject("odate").getString("hour");
            String sex = json.getString("sex");

            return new Tuple2<>(new TimeSex(hour, sex), 1);
        });

        for (int i = 0; i < 24; i++) {
            transAndPut(pairRDD, i+"");
        }
    }

    public static void transAndPut(JavaPairRDD<TimeSex, Integer> pairRDD, String hour) {
        JavaPairRDD<TimeSex, Integer> hourRDD = pairRDD.filter(t -> t._1.getHour().equals(hour));
        JavaPairRDD<TimeSex, Iterable<Integer>> groupBySex = hourRDD.groupByKey();

//        groupBySex.foreach(t -> System.out.println(t));

        JavaRDD<TimeSex> map = groupBySex.map(t -> {
            TimeSex timeSex = t._1;
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            if (timeSex.getSex().equals("男")) {
                timeSex.setMaleCount(count);
            } else if (timeSex.getSex().equals("女")) {
                timeSex.setFemaleCount(count);
            } else {
                throw new Exception("类型错误");
            }
            return timeSex;
        });
//        map.foreach(t -> System.out.println(t));

        List<TimeSex> take = map.take(2);
        TimeSex result = take.get(0);
        if (result.getSex().equals("男")) {
            result.setFemaleCount(take.get(1).getFemaleCount());
        } else {
            result.setMaleCount(take.get(1).getMaleCount());
        }
//        result.set(result.getViewCount()+result.getCartCount()+result.getParchaseCount());
        result.setMalePercent((float)result.getMaleCount() * 100 / (result.getMaleCount()+result.getFemaleCount()) + "%");
        result.setFemalePercent((float)result.getFemaleCount() * 100 / (result.getMaleCount()+result.getFemaleCount()) + "%");

        System.out.println(result);

        //put到HBase
        String[] columns = {"hour", "maleCount", "femaleCount", "malePercent", "femalePercent"};
        String[] values = {result.getHour(), result.getMaleCount()+"", result.getFemaleCount()+"",
        result.getMalePercent(), result.getFemalePercent()};
        try {
            HBaseUtils.putOneRowToHbase("time", result.getHour(), "sexinfo", columns, values);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
